package com.swak.netty.unicast;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

import com.swak.Constants;
import com.swak.OS;
import com.swak.exception.RemotingException;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandlerAdapter;
import com.swak.reactivex.transport.ExchangePeer;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.JsonMapper;
import com.swak.utils.Maps;

public class Node extends ExchangeHandlerAdapter {

	private LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 2, "Test.EventLoop-",
			false, 2, TimeUnit.SECONDS);

	ExchangePeer node;
	URL url;
	String name;

	public Node(String name) throws RemotingException {
		this.name = name;
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, Constants.ANYHOST);
		this.url = new URL("Test", OS.ip(), OS.randomAvailablePort(), parameters);
		this.node = Exchangers.peer(url, resources, this);
	}

	/**
	 * 非请求数据，才会执行这个回调
	 */
	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		System.out.println(name + " : 收到数据， 或者 oneway 的请求！" + JsonMapper.toJson(message));
	}

	/**
	 * 收到请求数据时，会执行这个并返回响应数据
	 */
	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		System.out.println("收到请求，直接返回请求");
		return CompletableFuture.completedFuture(message);
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		super.caught(channel, exception);
		if (exception != null) {
			exception.printStackTrace();
		}
	}

	public InetSocketAddress address() {
		return this.node.localAddress();
	}

	public void send(InetSocketAddress address, String message) throws RemotingException {
		this.node.send(Message.forMessage().setData(message).setAddress(address));
	}

	public static void main(String[] args) throws RemotingException {
		Node nodeA = new Node("A");
		Node nodeB = new Node("B");
		nodeA.send(nodeB.address(), "Hello B!");
		nodeA.send(nodeA.address(), "Hello A!");
	}
}
